package com.lili.collection.mq.runner;

import com.lili.collection.core.context.Context;
import com.lili.collection.core.runner.StageGroupRunner;
import com.lili.collection.core.runner.StageRunner;
import com.lili.collection.core.runner.pipe.AsyncPipeRunner;
import com.lili.collection.mq.group.AsyncProcessorGroupRunner;
import lombok.extern.slf4j.Slf4j;

/**
 *描述: MQ消息队列异步管道
 *@date:2023/3/17  19:20
 *@author:lili
 */
@Slf4j
public class MQAsyncPipeRunner extends AsyncPipeRunner {

    public MQAsyncPipeRunner(Context context) {
        super(context);
    }

    @Override
    public StageRunner getStageInputRunner() {
        return new AsyncInputRunner(getPipeContext());
    }

    @Override
    public StageRunner getStageOutputRunner() {
        return new AsyncOutputRunner(getPipeContext());
    }

    @Override
    public StageGroupRunner getStageProcessorGroupRunner() {
        return new AsyncProcessorGroupRunner(getPipeContext());
    }

    @Override
    public String getOutputOffset() {
        try {
            this.ensureStageRunner();
            return this.getOutputRunner().queryInitialOffset();
        } catch (Throwable throwable) {
            log.error("获取偏移量错误!",throwable);
        }
        return null;
    }
}